<?php
namespace Handler;

use App\Library\Log;
use App\Library\TextLog;
use App\Logger;
use Core\DzInternetProtocol;
use Core\CommandProtocol;
use Core\TanbaSwooleTable;
use Core\Model;
use Core\TbRedis;
use Helper\Log as stdLog;
use Swoole\Server;

class TanBa
{
    //token
    protected $token;
    protected $taskProcess;
    protected $cpu_num;


    //协议解析对象
    protected $command_protocol;

    //协议生成对象
    protected $build_protocol;

    //数据库操作类
    protected $db;
    protected $workerdb=array();

    //探霸链接列表
    protected $client;
    protected $heartlist;
    protected $ouibase;


    //redis连接句柄
    protected $redis;
    protected $wkrRedis=array();

    //监听端口访问列表
    protected $listener;

    //系统配置列表--同步更新数据库--比如rf设备上报数据间隔
    protected $sysconfig;


    /**
     * 构造函数
     */
    public function __construct()
    {
        try {
            $this->cpu_num = CPU_NUM; //swoole_cpu_num()*4;
            $this->protocol = new DzInternetProtocol();
            $this->command_protocol=new CommandProtocol();

            $swoole_table= new TanbaSwooleTable() ;
            $this->client =$swoole_table->client;
            $this->heartlist =$swoole_table->heartlist;
            $this->tanbalist =$swoole_table->tanbalist;

            $this->flow = new \Swoole\Table(4096);
            $this->flow->column('fd', \Swoole\Table::TYPE_INT, 8);
            //$this->flow->column('send', \Swoole\Table::TYPE_INT, 8); // 已发送数据流量
            //$this->flow->column('receive', \Swoole\Table::TYPE_INT, 12); // 已接收数据流量
            //$this->flow->column('failed', \Swoole\Table::TYPE_INT, 4); // 解析失败数据流量
            $this->flow->create();

            $this->listener =$swoole_table->listener;
            $this->ouibase =$swoole_table->ouibase;

            //创建客户端连接内存表
            $this->id2fd = new  \Swoole\table(1024);
            $this->id2fd->column('fd', \Swoole\table::TYPE_INT, 8);
            $this->id2fd->create();

            $this->swoole_serialize = new \Swoole\Serialize();

        } catch (Exception $e) {
            echo $e->getMesasge();
        }

    }

    /**
     * 获取探霸KEY
     * @param $fd
     * @return mixed
     */
    public function getLoggerInfo($fd)
    {
        return [
            'tbMac' => $this->client->get($fd, 'deviceID'),
            'method' => 'dataLog',
            'type'  => 'data',
        ];
    }

    public function getFlow()
    {
        return $this->flow;
    }

    /**
     * 主进程Master进程启动回调方法    初始化配置参数
     * @param  SwooleServer $serv    swoole对象
     */
    public function onStart($serv)
    {
        echo "TBserver config Master start \n";
        $thismstdb=Model::init(PRERDS."tanbaserverdatainit");
        $thismstredis=TbRedis::init(PRERDS."tanbaserverdatainit");

        $datas = $thismstdb->select('oui',  ["id", "Assignment","OrganizationName"]);
        if(!empty($datas)){
            foreach($datas as $item)
            {
                $this->ouibase->set(strtoupper($item['Assignment']), array('uid' => $item['id'],'ouiname'=>$item['OrganizationName']));
            }

            if($this->ouibase->exist('E043DB'))
                echo "\n DB [".config('db')['server']."] , oui database is ready ok.\n";
            else
                echo "\n DB [".config('db')['server']."]  swooletable ouibaseis ERR! \n";

        }else
            exit("\n DB mysql server [".config('db')['server']."] is ERR OR table oui is null! \n");

        foreach ($thismstredis->keys('DZ*') as $ky) {
                 $thismstredis->hMset($ky,array('status'=>0,  'fd' =>0,'updatetime' => time()));
        }

        foreach ($thismstredis->keys(date('Ymd', strtotime("-1 day")) . 'LOG_') as $ky) {
            $thismstredis->del($ky);
        }

    }
    // 2. Master进程拉起Manager进程
    public function onManagerStart( $serv )
    {
        cli_set_process_title(PRERDS."tanba_TbServer_data_manager");
        echo "\n TbServer Manager Start ...master_pid=>".$serv->master_pid.",manager_pid=>".$serv->manager_pid."\n";
        //   $serv->locProcess::setaffinity(array $cpu_set);//表示绑定哪些CPU核，如array(0,2,3)表示绑定CPU0/CPU2/CPU3

        //处理已过期的redis健值
        $this->expiredKeyProcess=new \Swoole\Process(function($proc) use($serv) {
            echo "\n - deal expired redis key Process is running -- \n";
            cli_set_process_title(PRERDS."basedataserver_delexpired_rediskey_Process");
            while (1) {
                try {
                    if (false ==  $theRedis = @TbRedis::init(PRERDS.'deal_expired_rediskey')) {
                        echo "Redis connect faild! \n";
                        throw new \Exception("Redis connect faild");
                    }

                    break;
                } catch (\Exception $e) {
                    echo "reconnect Redis …… \n";
                    sleep(2);
                }
            }
            while (1) {
                try {
                    if (false ==$theDb= @Model::init(PRERDS.'deal_expired_rediskey')) {
                        echo "Db connect faild! \n";
                        throw new \Exception("DB connect faild");
                    }

                    break;
                } catch (\Exception $e) {
                    echo "reconnect Db …… \n";
                    sleep(2);
                }
            }
            while ($data=$proc->read()){
                if(strpos($data,'dtheart:')){
                    list($prefix, $fd) = explode(':', $data);

                    if ($prefix !== 'dtheart') continue;

                    // 如果数据中可以读取出客户端连接编号，且该客户端处于连接状态， 关闭
                    if ($fd && $serv->exist($fd)) {
                       echo "\n ++ close client {$fd}, too long no heart recv! ----------- " . date('Y-m-d H:i:s') . "\n";

                        /**
                         * 心跳不存在
                         */
                       Log::dataLog($this->client->get($fd, 'deviceID'))->alert(Log::HEART_LISTEN, [
                           'fd'     => $fd,
                           '心跳异常' => "服务器长时间没有收到心跳"
                       ]);

                       $this->client->set($fd, array('status' => Logger::CLOSED_VIA_HEART_DIE));
                       $serv->close($fd);
                    }
                }


                if(strpos($data, PRERDS) == false ) continue;//非本地redis的DB库
                setExpiredKey($data,$theRedis,$theDb);
            }

        });
        $this->expiredKeyProcess->start();

        //监听检查过期的redis健值，清理redis
        (new \Swoole\Process(function ($process){
            cli_set_process_title('basedataserver_listen_expired_redis_key_Process');

            while (1) {
                try {
                    if (false ==  $prRedis = @TbRedis::init(PRERDS.'listenexpiredProcess')) {
                        echo "Redis connect faild! \n";
                        throw new \Exception("Redis connect faild");
                    }

                    break;
                } catch (\Exception $e) {
                    echo "reconnect Redis …… \n";
                    sleep(2);
                }
            }

            try{
                if(isset($prRedis) ){

                    $prRedis->punsubscribe(['__keyevent@'.config('redis')['db'].'__:expired']);

                    $pipe = $prRedis->multi();
                    //清除过期通知的健  DevTbExpired
                    foreach ($prRedis->keys('*.*Expired'.PRERDS) as $ky){
                        $pipe->del($ky);
                    }

                    foreach ($prRedis->keys('DZ*') as $ky){
                        if(! strpos($ky, ".")){
                            $pipe->hMset($ky, array(
                                'a000.count5'=>0,
                                'a001.count5'=>0,
                                'a002.count5'=>0,
                            ));
                        }
                    }

                    $ret=$pipe->exec();

                    echo "\n -- basedataserver listen expired redis_key Process  start……";

                    $prRedis->setOption(\Redis::OPT_READ_TIMEOUT, -1);
                    $prRedis->psubscribe(['__keyevent@'.config('redis')['db'].'__:expired'], function($redis, $chan, $chanName,$data) {
                        $this->expiredKeyProcess->write($data);
                    });
                }

            }catch (\ErrorException $e){
                echo $e->getMessage();
            }

        }, false, false))->start();

        //读取redis命令进程
        ( new \Swoole\Process(function ($process) use($serv) {

            while (true) {
                while (true) {
                    try {
                        $redis =TbRedis::init("dtsvrwbsctcommand");
                        break;
                    } catch (\RedisException $e) {
                        sleep(2);
                    }
                }

                while (true) {
                    try {
                        $data = json_decode($redis->brPop('dtsvrcommand', 0)[1], true);
                        if (isset($data['fd']) && isset($data['command']) ) {
                            $fd=$data['fd'];
                            if($serv->exist($fd)){
                                if($data['command']=='closeclient'){
                                    $this->client->set($fd, array('status' => 6));
                                    $serv->close($fd);
                                }elseif($data['command']=='resetRfugt'){
                                    $this->sendtoclient($serv,
	                                    $data['fd'],
	                                    $this->protocol->setCommand(
	                                    	['dt'=>'10','sdt'=>'0001'],
		                                    ['rfid'=>$data['tagID'],'do'=>'Rfcancel'],
	                                        $this->client->get($fd, 'token'))
                                    );
//                                    $cmdmsg="\n redis  command :{$data['command']}".json_encode($data)." \n";
//                                    echo  $cmdmsg;
//                                    echoMsg($cmdmsg,$redis,true);
                                }
                            }
                        }

                    } catch (\RedisException $e) {
                        echoMsg('redis  dataserver commander error [' . $e->getMessage() . ']', $e->getCode(), $e->getLine(),$redis);
                        break;
                    }

                }
            }
        }, false))->start();

    /*    //心跳发送进程

        (new \Swoole\Process(function ($process) use($serv) {
            cli_set_process_title('dataserver_heart_senter_Process');
            while (1) {
                sleep(3);
                foreach ($this->heartlist as $row) {
                    $fd=$row['fd'];
                    if ($serv->exist($fd)) {
                        if(isset($row['token']) && isset($row['tbmac']) && isset($row['sendtime']) && isset($row['timer'])){
                             $hrtstatus=$this->heartlist->get($fd, 'hrtstatus');
                             if(((time()-$row['sendtime'])>=$row['timer']*60 )|| $hrtstatus==1){

                                 if($hrtstatus==1){
                                     $errcounts=$this->heartlist->incr($fd,'senderro', 1);//心跳发送数量
                                     if($errcounts>3){
                                         $this->client->set($fd, array('status' => 7));
                                         $serv->close($fd);
                                         continue;
                                     }

                                 }

                                $token=$this->heartlist->get($fd, 'token');
                                $ret =  $serv->send($fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'heart'],$token));

                                $this->heartlist->set($fd,['hrtstatus'=>1,'sendtime'=>time()]);
                                if($ret){
                                    $this->client->set($fd,['hrtstatus'=>1]);//发送标记hrtstatus，收到响应后取消
                                    $this->client->incr($fd,'hrtsd', 1);//心跳发送数量

                                    echo "\n +++Heart! [{$row['tbmac']}] ".date('H:i:s',$row['sendtime'])." ++++ \n";
                                }
                            }

                        }

                    }

                }
            }


        }))->start();
*/

        // 每天清理前天的探霸数据
        /*(new \Swoole\Process(function () {
            while (1) {


            }
        }))->start();*/


        /**
         * @mark lvzh
         * TODO 增加一个临时token， 防止token异步问题
         * 每隔开一小时处理TOKEN
         */
        (new \Swoole\Process(function ($process) use($serv) {
            cli_set_process_title('dataserver_token_sent_Process');
            while (1) {
                // sleep(TOKEN_INTERVAL);
                sleep(CHECK_TOKEN_INTERVAL);
                foreach ($this->client as $row) {
                    $fd=$row['fd'];
                    if ($serv->exist($fd) && isset($row['token'])) {

                        /*dump('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>');
                        dump([
                            'fd' => $fd,
                            'client' => $this->client->get($fd, 'deviceID')
                        ]);*/
                        // 拿出token， 对比当前时间， 将超时的token替换
                        $oldToken = $row['token'];
                        $newToken = time();

                        if ($newToken - $oldToken < TOKEN_INTERVAL) continue;
                        //if ($newToken - $oldToken < 2 * 60) continue;

                       /* dump([
                            'fd' => $fd,
                            'client' => $this->client->get($fd, 'deviceID')
                        ]);
                        dump('<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<');*/

                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'sdtoken','token'=>$newToken],$oldToken));
                        echo "sent [{$row['deviceID']}] token!-{$newToken}";

                        Log::dataLog($this->client->get($fd, 'deviceID'))->debug('TOKEN更换', [
                            'fd' => $fd,
                            'oldToken' => dechex($oldToken),
                            'newToken' => dechex($newToken)
                        ]);


                        //$this->client->set($fd, array('token' =>$newtoken,'newtoken' =>$newtoken,'hrtsderr'=> 0,'openheart'=>0));
                        $this->client->set($fd, array('newtoken' =>$newToken,'hrtsderr'=> 0,'openheart'=>0));
                        $this->heartlist->set($fd,['token'=>$newToken,'status'=>0,'senderro'=>0]);

                        /* $newtoken=time();
                        $token=$row['token'];
                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'sdtoken','token'=>$newtoken],$token));
                        echo "sent [{$row['deviceID']}] token!-{$newtoken}";

                        //$this->client->set($fd, array('token' =>$newtoken,'newtoken' =>$newtoken,'hrtsderr'=> 0,'openheart'=>0));
                        $this->client->set($fd, array('newtoken' =>$newtoken,'hrtsderr'=> 0,'openheart'=>0));
                        $this->heartlist->set($fd,['token'=>$newtoken,'status'=>0,'senderro'=>0]);*/
                    }

                }
            }

        }))->start();


    }

    /**
     * worker进程启动回调方法
     * @param  SwooleServer $serv    swoole对象
     * @param  int $worker_id         worker进程id号
     */
    public function onWorkerStart($serv, $worker_id)
    {
        //  //  $serv->tick(1000, function ($id) {        var_dump($id);    });
        //连接数据库
        while (true) {
            try {
                $serv->db[$worker_id] = Model::init(PRERDS."dataswrwkr1".$worker_id);
                break;
            } catch (\PDOException $e) {
                stdLog::print('mysql connect fail on worker['.$serv->worker_id.']', $e->getCode(), $e->getLine());
            }
        }

        //连接redis
        while (true) {
            try {
                $serv->redis[$worker_id] =TbRedis::init(PRERDS."datarsv1.".$worker_id);
                break;
            } catch (\PDOException $e) {
                stdLog::print('redis connect fail on worker['.$serv->worker_id.']', $e->getCode(), $e->getLine());
            }
        }

    }

    /**
     * 连接回调方法
     * @param  SwooleServer $serv    swoole对象
     * @param  int $fd                连接句柄编号
     */
    public function onConnect($serv, $fd)
    {
        $info = $serv->connection_info($fd);
        $theredis= $serv->redis[$serv->worker_id];

        $token = time();
        $this->client->set($fd, array('token' =>$token, 'newtoken' =>$token,'fd' => $fd,'lasttime'=>time(),'firsttime'=>microtime_float()));
        $this->heartlist->set($fd,['fd'=>$fd,'intime'=>$token,'sendtime'=>$token,'timer'=>HEART_INTERVAL,'hrtstatus'=>0,'senderro'=>0]);

        // @test 流量设置初始化
        //$this->flow->set($fd, ['fd' => $fd, 'send' => 0, 'receive' => 0, 'failed' => 0]); // 流量设置初始化

        $msg= "\n ==== onConnect :worker_id=".$serv->worker_id.",fd=$fd,from=".$info['remote_ip'].":".$info['remote_port'] ." : ". date('Y-m-d H:i:s') ." ====\n";
        echo $msg;
        echoMsg($msg,$theredis,true);


    }

    /**
     * 接收数据回调方法
     * @param  \Swoole\Server $serv    swoole对象
     * @param  int $fd                连接句柄编号
     * @param  int $form_id           reactor线程号
     * @param  bin $data              接收到的二进制数据
     */
    public function onReceive($serv, $fd, $form_id, $data)
    {
        $info = $serv->connection_info($fd);
        /**
         * @var \Redis $theRedis
         */
        $theRedis=$serv->redis[$serv->worker_id];
        $theDb=$serv->db[$serv->worker_id];

        $hexStr=bin2hex($data);
        if(!empty($hexStr)){
            $this->client->incr($fd,'receivects', strlen($hexStr)/2); //字节数
            $this->client->incr($fd,'tcount', 1);
        }

        $dataBytes = strlen($hexStr) / 2; // 收到数据的字节数
        //$this->flow->incr($fd, 'receive', $dataBytes);


        // 连接所耗费的流量忽略
        // @test 收到流量计算, 连接所耗费的流量忽略
        if ($deviceID = $this->client->get($fd, 'deviceID')) {
            $tbKey = Logger::tb($deviceID);
            $receiveBytes = $theRedis->hGet($tbKey, 'dataReceiveBytes') ?: 0;

            $receiveBytes += $dataBytes;

            // 简化单词数量
            TextLog::dataLog($deviceID)->debug('recv:', [
                'cur' => $dataBytes . 'B',
                'all' => number_format($receiveBytes / 1024, 3) . 'KB',
            ]);

            $theRedis->hSet($tbKey, 'dataReceiveBytes', $receiveBytes);
        }
        $token = $this->client->get($fd, 'token');

        $nowtime=time();
//
//        $time0=$this->client->get($fd,'firsttime');
//        echo "\n --0-connect to receive first request info  times :-----".$fd."||".((microtime_float()-$time0)/1000).PHP_EOL;
        $data = $this->protocol->parsein($hexStr,$token);    //解析数据

      /*  'header'=>$data['header'],
            'option'=>$data['option'],
            'data'=>[
        'item1'=>'A1020008'.$data['option']['rfid']
    ],
            'token'=>$this->token*/

        //\SeasstdLog::debug('ok?');
        if(isset($data['pserro'])){
            $tbmac=$this->client->get($fd,'deviceID');
            //echo "Erro-".$data['pserro']."({$fd}[".($tbmac?:'')."])".PHP_EOL;

            // 如果存在newToken重新解析
            if ($newToken = $this->client->get($fd, 'newtoken')) {
                $data = $this->protocol->parsein($hexStr, $newToken);    //解析数据

                /**
                 * 记录旧换新token
                 */
                Log::dataLog($tbmac)->debug(Log::TOKEN_LISTEN, [
                    'fd' => $fd,
                    'oldTokenError' => "使用新token重新解析数据",
                    'data' => $hexStr,
                    'oldToken' => dechex($token)
                ]);

                if (isset($data['pserro'])) {
                    $this->client->incr($fd, 'invalid', 1);
                    // 错误数据，强制断开
                    //dump('ERROR start------------>>>>>>>>>>>>');
                    //dump("Error-{$data['pserro']}(fd:{$fd}[tb:{$tbmac}])");
                    /**
                     * token校验失败
                     */
                    Log::dataLog($tbmac)->alert(Log::TOKEN_LISTEN, [
                        'fd' => $fd,
                        'newTokenError' =>  "新token解析数据依旧失败",
                        'newToken' => dechex($newToken),
                    ]);
                    $this->client->set($fd, array('status' => Logger::CLOSED_VIA_TOKEN_ERROR));

                    $this->flow->incr($fd, 'failed', $dataBytes);
                    $serv->close($fd);
                   // dump('ERROR end--------<<<<<<<<<<<<<<<<<');
                    return false;
                } else {
                    // 更换token
                    //\SeasstdLog::info("TokenUpdate-fd:{$fd}[tb:{$tbmac}]");
                    $this->client->set($fd, [
                        'token' => $newToken
                    ]);
                }
            } else {
                $this->client->incr($fd, 'invalid', 1);
                // 错误数据，强制断开
                //dump('ERROR start------------>>>>>>>>>>>>');
                //dump("Error-{$data['pserro']}(fd:{$fd}[tb:{$tbmac}])");
                //\SeasstdLog::debug("Error-{$data['pserro']}(fd:{$fd}[tb:{$tbmac}])");
                Log::dataLog($tbmac)->emergency(Log::UNKNOWN_ERROR, [
                    'fd' => $fd,
                    '解析错误' => "无法正确解析数据",
                    '原始数据' => $data,
                ]);
                $this->client->set($fd, array('status' => Logger::CLOSED_VIA_UNKNOWN_DATA));
                $this->flow->incr($fd, 'failed', $dataBytes);
                $serv->close($fd);
                //dump('ERROR end--------<<<<<<<<<<<<<<<<<');
                return false;
            }
        }

        if(!$data || !isset($data['data']) || !isset($data['header'])){
            $this->client->incr($fd,'othercts', 1); //其他无效数据
            // 错误数据，强制断开
            //dump('UNKNOWN start-------->>>>>>>>>>>>>');
            //dump('other error!');
            Log::dataLog($this->client->get($fd,'deviceID'))->emergency(Log::UNKNOWN_ERROR, [
                'fd' => $fd,
                '解析错误' => "无法正确解析数据",
                '原始数据' => $data,
            ]);
            $this->client->set($fd, array('status' => Logger::CLOSED_VIA_UNKNOWN_DATA));
            $this->flow->incr($fd, 'failed', $dataBytes);
            $serv->close($fd);
            //dump('UNKNOWN end-------->>>>>>>>>>>>>');
            return false;
        }


        if($data['header']['dt']=='00'){
            switch ($data['header']['sdt']) {
                case '0011': // 连接请求帧
                    //var_dump($data['data']);
                    if(isset($data['data']['encodeType'])){
                        $this->client->set($fd, array('encodetype' =>$data['data']['encodeType']['code'],'checktype'=>$data['data']['encodeType']['check']));
                    }
                    if(isset($data['data']['device'])){
                        $tanbaMac=$data['data']['device']['mac'];
                        $time0=$this->client->get($fd,'firsttime');

                        /**
                         * receive connect info cost times
                         * 探霸从连接服务器到收到连接请求帧一共花费了多少时间
                         *
                         */
                        $msg= "\n ==== onConnect ({$fd})[{$tanbaMac}]: receive connect info times :".((microtime_float()-$time0)/1000)." s ===== ".PHP_EOL;
                        echo $msg;
                        echoMsg($msg,$theRedis,true);
                        // 统计当天上线次数
                        $tbMacKey = Logger::tb($tanbaMac);
                        $connectCounts = $theRedis->hGet($tbMacKey, 'connect_counts') ?: 0;
                        $theRedis->hMSet($tbMacKey, [
                            'connect_counts' => ++$connectCounts,
                            'connect_at'    => time()
                        ]);

                        Log::dataLog($tanbaMac)->info(Log::MANAGER_FRAME, [
                            'fd' => $fd,
                            'onConnect' => '连接花费时间为: ' . ((microtime_float()-$time0)/1000) . 's',
                            '连接次数'    => $connectCounts
                        ]);

                        /**
                         * 每次上线进行一次记录
                         */
                        // 获取当前上线次数


                        //如果连接已经存在则关闭上一个连接
                        if ($this->id2fd->exist($tanbaMac)) {
                            $oldLink = $this->id2fd->get($tanbaMac, 'fd');
                            if (isset($oldLink) && !empty($oldLink) && $oldLink != $fd) {
                                $this->client->del($oldLink);
                                $this->client->set($oldLink, array('status' => Logger::CLOSED_VIA_LINK_TWICE));
                                $serv->close($oldLink);
                            }
                        }

                        $theRedis->hMset($tanbaMac, array(
                            'encodetype'=>$data['data']['encodeType']['code'],
                            'checktype'=>$data['data']['encodeType']['check'],
                            'basedeviceinfo' => json_encode($data['data']['device'])
                        ));

                        //记录版本、类型和mac标识
                        $this->client->set($fd, array('deviceID' =>$tanbaMac,
                            'devicetype'=>$data['data']['device']['type'],
                            'version'=>$data['header']['vs'],
                            'encodetype' =>$data['data']['encodeType']['code'],
                            'checktype'=>$data['data']['encodeType']['check']

                    ));
                        $this->id2fd->set($tanbaMac, array('fd' => $fd));

                        //回应连接段连接请求响应
                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'00','sdt'=>'0100']));
                        sleep(1);

                        //发送token
                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'sdtoken','one'=>0],$token));
	                    $time0=$this->client->get($fd,'firsttime');
	                    echo "\n --2-connect to send token times :-----".$fd."|".$tanbaMac."|".((microtime_float()-$time0)/1000).PHP_EOL;
	
	                    Log::dataLog($tanbaMac)->info('初始化发送TOKEN', [
		                    'fd' => $fd,
		                    'token' => dechex($token),
	                    ]);


                        sleep(1);
                        $serv->send($fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'heart'],$token));
                        $this->heartlist->set($fd,['fd'=>$fd,'tbmac'=>$tanbaMac,'sendtime'=>$nowtime,'token'=>$token,'timer'=>HEART_INTERVAL]);


                        sleep(1);
                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'0110'],['do'=>'getdeviceinfo'],$token));

                        //$serv->send($fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'0110'],['do'=>'getdevicestatus'],$token));

                        // 获取心跳间隔 ok
                        sleep(1);
                        $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'getheart'],$token));

                        $deviceID=$tanbaMac;
                        $tanbaid = $theRedis->hGet($deviceID, 'tbid');
                        if(!isset($tanbaid) || empty($tanbaid)) {

                            //     //重新设置 插入或者更新数据-------------------
                            $insert_data = array(
                                'mac' => $deviceID,
                                'token' => $this->client->get($fd, 'token'),
                                'serverip' => IP,
                                'serverport' => PORT,
                                'datafd' => $fd,
                                'status' => 1,
                                'lasttime' => time(),
                                'createtime' => time(),
                            );
                            $where = array('mac' => $deviceID);
                            if (!$theDb->has('device', $where)) {
                                $tanbaid = $theDb->insert('device', $insert_data);
                                $tbmsg = "\n new TBdevice [" . $deviceID . "] insert successful \n";
                                echoMsg($tbmsg, $theRedis, true);
                            }else{
                                $this->sentmysqltask($serv,'device', $insert_data, $where,'update');
                            }

                            $tanbaid = $theDb->get('device', 'id', $where);

                        }

                        if (!$theDb->has('device', ["id" => $tanbaid])) {
                            $theDb->insert("device_basehome", ["dvid" => $tanbaid, "nickname" => $deviceID]);
                        }

                        $serv->send($fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'0110'],['do'=>'getdeviceloc'],$token));

                        $this->client->set($fd, array('tbid' => $tanbaid));
                        $this->tanbalist->set($deviceID, array("fd" => $fd, 'tbid' => $tanbaid, "lasttime" => time()));

                        //设置缓存
                        $tbloction = $theDb->get("device_basehome", ["longitude", "latitude", 'hightland'], array("dvid" => $tanbaid));
                        $array_hmset_date = array(
                            'fd' => $fd,
                            'id' => $tanbaid,
                            'mac' => $deviceID,
                            'tbid' => $tanbaid,
                            'encodetype'=>$data['data']['encodeType']['code'],
                            'checktype'=>$data['data']['encodeType']['check'],
                            'basedeviceinfo' => json_encode($data['data']['device']),
                            'location' => json_encode($tbloction?$tbloction:[])
                        );
                        $theRedis->hMset($deviceID, $array_hmset_date);

                        checktbold($deviceID,$theRedis);

                        $sttdata= array(
                            'type' => 'TBstat',
                            'fd' =>$fd,
                            'deviceid' => $deviceID,
                            'status' => 1,
                            'statustext' => 'online',
                            'time' => date('Y-m-d H:i:s',time()),
                        );
                        $this->sentMsgOut($serv,$sttdata,$theRedis);

                        //记录日志
                        $data = array(
                            'type' => 0,
                            'deviceid' => $deviceID,
                            'description' => '设备连接成功',
                            'ip' => $info['remote_ip'],
                            'createtime' => time(),
                        );
                        $this->sentmysqltask($serv, 'devicelog', $data);

                    }

                    break;
                case '0111':
                    //客户端请求断开--探霸可暂时不用
                    echo "=== close request 0111 ===\n";
//                    var_dump($data['data']);
//                    $serv->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'00','sdt'=>'0111','td'=>'10'],['do'=>'setcloseresponse','data'=>1],$token));//同意断开

                    break;
                case '1000':
                    //服务器要求断开客户端，然后客户端的回应，要求断开链接
                    echo "=== close request 1000 ===\n";
//                    var_dump($data['data']);
                    break;

                default:
                    ;
            }
        }elseif ($data['header']['dt']=='01'){
            if(isset($data['data']['fromcmd']) ){
                $deviceID=$this->client->get($fd, 'deviceID');
                $tanbaid=$this->client->get($fd, 'tbid');

                echo PHP_EOL."({$fd})[{$deviceID}] return:".$data['data']['fromcmd'].PHP_EOL;
                switch ($data['data']['fromcmd']) {
                    case 'heart':
                        echo "\n {$deviceID} @";
                        $this->heartlist->set($fd,['resvtime'=>time(),'status'=>0,'hrtstatus'=>0,'senderro'=>0]);

                        $this->client->set($fd,['hrtstatus'=>0]);
                        $this->client->incr($fd,'hrtrsv', 1);//收到的数量
                        $sttdata= array(
                            'type' => 'TBstat',
                            'fd' =>$fd,
                            'deviceid' => $deviceID,
                            'status' => 1,
                            'statustext' => 'online',
                            'time' => date('Y-m-d H:i:s',time()),
                        );
                        $this->sentMsgOut($serv,$sttdata,$theRedis);


                        $theRedis->set('dtheart:' . $fd, 1);
//                        $timer = $theRedis->hGet($deviceID, 'datatimer');
                        $theRedis->expire('dtheart:' . $fd,  2 * 60);

                        echo "\n ------------++ recv client {$fd}【{$deviceID}】's heart! ----------- " . date('Y-m-d H:i:s') . "|timer  2 min--------\n";

                        $serv->send($fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'respheart'],$token));
/*                        dump('>>>error');
                        dump($this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'respheart'],$token));
                        dump('<<<!!!');*/
                        Logger::dataLog($deviceID)->debug("heart| ------------++ recv client {$fd}【{$deviceID}】's heart! ----------- " . date('Y-m-d H:i:s') . "---timer  2 min--------");
                        break;
                    case 'getdeviceinfo':
                       //var_dump($data['data']);
                        echo "*********************getdeviceinfo*******************\n";
                        //插入或者更新数据库
//                            $rdata['device']['wktype'];//工作模式
//                            $rdata['device']['vltype'];//协议版本
//                            $rdata['device']['hdtype'];//硬件版本
//                            $rdata['device']['sttype'];//软件版本

                        break;
                    case 'getdeviceloc':
//                            var_dump($data['data']['device']);

                        if(isset($data['data']['location']['longitude']) && !empty($data['data']['location']['longitude']) && $data['data']['location']['latitude']){
                            $tbloction=array("longitude"=>$data['data']['location']['longitude'], "latitude"=>$data['data']['location']['latitude'], "hightland"=>$data['data']['location']['hightland']);
                            $theDb->update("device_basehome",$tbloction, ["dvid" => $tanbaid, "nickname" => $deviceID]);

                            $theRedis->hMset($deviceID, array('location' => json_encode($tbloction)));
                        }
                        break;
                    case 'getability':
                        file_put_contents('/var/www/2.txt', print_r($data['data'], true), FILE_APPEND);
                        break;
                    case 'setheart':
                        if($data['data']=='成功'){
                            $this->sendtoclient($serv,$fd, $this->protocol->setCommand(['dt'=>'01','sdt'=>'1000'],['do'=>'getheart'],$token));
                        }
                        break;
                    case 'getheart':
                        if($data['data']['data']>0){
                            $this->sentmysqltask($serv,'device', ['dataheart'=>$data['data']['data']], array('mac' => $deviceID),'update');
                            $this->heartlist->set($fd,['timer'=>$data['data']['data']]);

                            $theRedis->hMset($deviceID, ['datatimer'=>$data['data']['data']]);
                        }

                        break;
                    default:

                        //var_dump($data['data']);
                        break;
                }

            }else{
                $tbmac=$this->client->get($fd,'deviceID');
                $htmsg= "\n Erro:=== ({$fd})[{$tbmac}]receive erro ,decode type  not find :{$hexStr} ===\n";
                //echo $htmsg;
//dump($data);
                echoMsg($htmsg,$theRedis,true);
                Log::dataLog($tbmac)->alert(Log::CONTROL_FRAME, [
                    'fd'       => $fd,
                    '帧类型错误' => "没有对应的控制帧类型",
                    '原始数据'  => $hexStr
                ]);
            }
        }
        elseif ($data['header']['dt']=='10'){
            $tanbaMac=$this->client->get($fd,'deviceID');
            $tbid=$this->client->get($fd,'tbid');

            /*if ($tanbaMac == 'DZ180809010000002Q' ) {
                //if (isset($data['data']['type']))
                    dump($data);
            }*/

            if(!is_array($data['data'])){
                $htmsg= "\n Erro:=== ({$fd})[{$tanbaMac}]receive !is_array(data) :{$hexStr} ===\n";
                echo $htmsg;
                echoMsg($htmsg,$theRedis,true);

                Log::dataLog($tanbaMac)->alert(Log::DATA_FRAME, [
                    'fd' => $fd,
                    '数据错误' => "收到的数据解析后不是一个数组",
                    '原始数据' => $hexStr
                ]);

                $this->sentMsgtoTcp($serv, ['tbid'=>$tbid,'tbmac'=>$tanbaMac,'data'=>json_encode($data),'type'=>'b000', /*'tbid' =>0,*/'mac'=>0,'rssi'=>0,
                'devkey' =>'0', 'devid' =>0,'find'=>0,'ApMac'=>0,'dvuid'=>0, 'time' => time() ]);

            }else{
                if(isset($data['data']['type']) && $data['data']['type']=='typeErro'){
                    echo " \n ======={$fd}[{$tanbaMac}]=={$hexStr}|{$token}====other type notin a000a001a002: ".json_encode($data['data'])."================".PHP_EOL;

                    Log::dataLog($tanbaMac)->alert(Log::DATA_FRAME, [
                        'fd'     => $fd,
                        '类型错误' => '类型非a000, a001, a002',
                        'token'  => $token,
                        '原始数据' => $hexStr,
                        '解析后的数据' => json_encode($data['data'])
                    ]);

                    $this->client->incr($fd,'Errotc', 1);
                }

                $nowtime=time();
                foreach ($data['data'] as $key => $item) {

                    if (!empty($item) && is_array($item)  && isset($item['type']) && in_array($item['type'], ['a000', 'a001', 'a002'])) {
                        $devkey=$item['mac'];
                       $type= $item['type'];
                        $dataitem=array_merge(array(
                            'tbid' =>$tbid,
                            'tbmac'=>$tanbaMac,
                            'devkey' =>$devkey,
                            'time' => $nowtime
                        ),$item);

//                            $dvtype=DzDataType($item['type']);
                        if($item['type']=='a001' ) {
                            $dvtype='STA';
                            $macfg = substr($devkey, 0, 6);
                            $uid=$this->ouibase->get(strtoupper($devkey),'uid');
                            $dv_brand=$this->ouibase->get(strtoupper($macfg),'ouiname');

                            $apmac = substr($item['ApMac'], 0, 6);
                            $apuid=$this->ouibase->get(strtoupper($apmac),'uid');
                            $ap_brand=$this->ouibase->get(strtoupper($apmac),'ouiname');

                            $dataitem=array_merge(array(
                                'dvuid' =>empty($uid) ? "0":$uid,
                                'apuid' =>empty($apuid) ? "0":$apuid,
                                'dv_brand'=> empty($dv_brand) ? "未知" : $dv_brand,
                                'ap_brand'=> empty($ap_brand) ? "未知" : $ap_brand
                            ),$dataitem);

                        }elseif($item['type']=='a002'){
                            if($item['urgentStatus'] == '1') {
                                $urfinfo= $theRedis->hGet("UrRF_".$devkey,'loginfo');
                                if($urfinfo){
                                    $rglog=  json_decode($urfinfo, true);
                                    $rglog=array_merge($rglog,array('tb.'.time()=>array( 'tbmac'=>$tanbaMac,'tbid'=>$tbid,'time' =>$nowtime)));
                                }else{
                                    $rglog=array('tbid.'.time()=>array('tbmac'=>$tanbaMac,'tbid'=>$tbid,'time' => $nowtime));
                                }

                                $theRedis->hMSet("UrRF_".$devkey,['urgentStatus'=>1,'loginfo'=>json_encode($rglog)]);
                                //注意写入数据库
                                //数据响应帧
                                $serv->send($fd, $this->protocol->setCommand(['dt'=>'10','sdt'=>'0001'],['rfid'=>$devkey,'do'=>'Rfcancel'],$token));

                            }

                        }

                        if (abs($dataitem['rssi'])>127)  continue;

                        $dataitem['devid'] = setNewDeviceInfo($dataitem, $theRedis, $theDb);

                        $dataitem['find'] = setDevCount($dataitem, $theRedis, $theDb,2);
                        echo "^";
                        $this->client->incr($fd,$type.'dc', 1);
//                        echo "^({$fd})"."|".$dataitem['type']."|".$dataitem['devkey']."|".$dataitem['tbid']."\n";

                        $this->sentMsgtoTcp($serv, $dataitem);

//                        //1秒计数和压缩
                        $snowtime=substr($nowtime,-1,3);
                        $sdevrkey ="secrar.". substr($dataitem['type'],-1) . "." . $dataitem['devkey'] . "." . $dataitem['tbid']. $snowtime;
                        $sfinds=$theRedis->HINCRBY($sdevrkey,'finds',1);
                        if($sfinds==0)
                            $sfinds=$theRedis->hset($sdevrkey,'finds',1);

                        $oldara=json_decode($theRedis->hget($sdevrkey,'datarar'),true);
                        if(!$oldara|| !is_array($oldara)) $oldara=[];
                        $item= array_remove($item,'mac');
                        $item= array_remove($item,'type');
                        $theRedis->hMset($sdevrkey,['datarar'=>json_encode(array_merge($oldara,[$sfinds=>$item])),'tempdata'=>json_encode($dataitem)]);

                        $rdsdata=$theRedis->hGetAll($sdevrkey);
                        if(!$rdsdata || !isset($rdsdata['afttimer']) || !isset($rdsdata['sectime'])){
                            $devmkeyscd=$serv->after(1000, function() use ($serv, $sdevrkey,$dataitem,$theRedis) {
                                if($theRedis->exists($sdevrkey)){
                                    $rdsdata=$theRedis->hGetAll($sdevrkey);
                                    $dessfind= $rdsdata['finds'];
                                    $dataitem['json']= $rdsdata;
                                    $dataitem['sfind']= $dessfind;
                                    $this->sentMsgtoTcp($serv, $dataitem,'getsecrar');
//                                    echo "\n".$sdevrkey."=[".time()."]={$dessfind}/1s ==*1=\n";
                                    $theRedis->del($sdevrkey);
                                }

                            });
                            $theRedis->hMset($sdevrkey,['afttimer'=>$devmkeyscd,'sectime'=>$dataitem['time']]);

                        }else{
                            if($sfinds>29){
                                if(isset($rdsdata['afttimer']) && $rdsdata['afttimer']){
//                                    swoole_timer_clear($rdsdata['afttimer']);
                                    $kltm=@$serv->clearTimer($rdsdata['afttimer']);
                                    unset($this->heart_timer[$rdsdata['afttimer']]);
                                }
                                $dessfind= $rdsdata['finds'];
                                $dataitem['json']= $rdsdata;
                                $dataitem['sfind']= $dessfind;
                                $this->sentMsgtoTcp($serv, $dataitem,'getsecrar');
//                                echo "\n".$sdevrkey."=[".time()."]={$dessfind}/1s ==*2=\n";
//TODO:: 过于频繁，应该预警或限制
                                $theRedis->del($sdevrkey);
                            }
                        }

                    }else{
                        $htmsg= "\n Erro:=== ({$fd})[{$tanbaMac}]receive erro ,decode type  not find :{$hexStr} ===\n";
                        echo $htmsg;
                        //dump($item);
                        echoMsg($htmsg,$theRedis,true);
                        Log::dataLog($tanbaMac)->alert(Log::DATA_FRAME, [
                            'fd' => $fd,
                            '帧类型错误' => '对应的数据帧类型不存在',
                            '原始数据' => $hexStr,
                        ]);
                    }//

                }
            }

        }
        $data=null;
        unset($data);

    }


    public function onTask($serv, $task_id, $from_id, $data )
    {
        $thetaskRedis=$serv->redis[$serv->worker_id];

        static $conn = null;
        if($conn == null){
            $conn =new Model();
            if(!$conn){
                $conn = null;
                echo "TbServer Swoole Task DB Connect ERR:\n";
                echoMsg( "TbServer Swoole Task DB Connect ERR:\n",$thetaskRedis,true);
                return;
            }
        }



        $sqldata=$this->swoole_serialize::unpack($data);
        if(empty($sqldata)) return ;
        //   $sqldata=json_decode($data,true);

        if(isset($sqldata['table'])&& is_array($sqldata['param']) ){
            if ($sqldata['option']=='insert'){
                $ret=  $conn->insert($sqldata['table'],$sqldata['param']);
            }
            elseif ($sqldata['option']=='update'&& is_array($sqldata['where']) ){
                $ret=  $conn->update($sqldata['table'],$sqldata['param'],$sqldata['where']);
            }

        }

    }


    public function onFinish($serv, $task_id, $data)
    {
        // 异步Task时有效
        echo $data; //"\n # {$task_id} Task finished, result:" .$data.PHP_EOL; //       echo "AsyncTask Finish:Connect.PID=".posix_getpid().PHP_EOL;//

    }


    /**
     * 关闭回调方法
     * @param  SwooleServer $serv    swoole对象
     * @param  int $fd                连接句柄编号
     */
    public function onClose($serv, $fd, $reactorId)
    {
        $theWorkerRedis=$serv->redis[$serv->worker_id];
        $theDb=$serv->db[$serv->worker_id];

        echo " \n *** tcp total connects count is :".count($serv->connections)." ***".PHP_EOL;

        $deviceID=$this->client->get($fd, 'deviceID');
        if(isset($deviceID) && !empty($deviceID))
        {
            $status= $this->client->get($fd,'status');
            $clsmsg = $status==0 ? '':getclosetext($status);
            $sttdata= array(
                'type' => 'TBstat',
                'fd' =>$fd,
                'deviceid' => $deviceID,
                'status' => 0,
                'statustext' => 'offline',
                'offtext'=>$clsmsg,
                'time' => date('Y-m-d H:i:s',time()),
            );
            $this->sentMsgOut($serv,$sttdata,$theWorkerRedis);
            $htmsg="\n ==={$clsmsg}=== ";

            // 统计当天上线次数
            $tbMacKey = Logger::tb($deviceID);
            $closedCounts = $theWorkerRedis->hMGet($tbMacKey, [
                'connect_closed',
                'closed_by_server',
                'closed_by_client'
            ]);

            $connectClosedCounts = $closedCounts['connect_closed'] ?: 0;
            $closedByServerCounts = $closedCounts['closed_by_server'] ?: 0;
            $closedByClientCounts = $closedCounts['closed_by_client'] ?: 0;

            ++$connectClosedCounts;

            if($reactorId < 0) {
                $htmsg .= "server fd({$fd}) [{$deviceID}]". date('Y-m-d H:i:s',time())." closed. =====";

                ++$closedByServerCounts;

                // 被服务器端强制中断
                Log::dataLog($deviceID)->warning(Log::CONNECT_CLOSED, [
                    'status' => $status,
                    '断开原因' => '服务器断开: ' . Logger::getCloseReason($status),
                    '断开次数' => "总共断开{$connectClosedCounts}次, 其中服务器主动断开{$closedByServerCounts}次",
                    'fd' => $fd,
                    //'当前连接总接收流量' => $this->flow->get($fd, 'receive') / 1024 . 'KB',
                    //'当前连接总发送流量' => $this->flow->get($fd, 'send') / 1024 . 'KB',
                    //'解析错误流量' => $this->flow->get($fd, 'failed') . 'B',
                ]);
            } else {
                $htmsg .= "client fd({$fd}) [{$deviceID}] ".  date('Y-m-d H:i:s',time())."  closed. =====";

                ++$closedByClientCounts;
                // 客户端关闭连接
                Log::dataLog($deviceID)->warning(Log::CONNECT_CLOSED, [
                    'status' => $status,
                    '断开原因' => '客户端断开: ' . Logger::getCloseReason($status),
                    '断开次数' => "总共断开{$connectClosedCounts}次, 其中客户端主动断开{$closedByClientCounts}次",
                    'fd' => $fd,
                    //'当前连接总接收流量' => $this->flow->get($fd, 'receive') / 1024 . 'KB',
                    //'当前连接总发送流量' => $this->flow->get($fd, 'send') / 1024 . 'KB',
                    //'解析错误流量' => $this->flow->get($fd, 'failed') . 'B',
                    '测试' => $this->client->get($fd, 'sentcts') / 1024 . 'KB',
                ]);
            }

            $theWorkerRedis->hMset($tbMacKey, [
                'connect_closed' => $connectClosedCounts,
                'closed_by_server' => $closedByServerCounts,
                'closed_by_client' => $closedByClientCounts
            ]);

            $htmsg .=  $this->getpackmsg($fd);
            Log::dataLog($deviceID)->warning(Log::CONNECT_CLOSED, [
                $htmsg
            ]);
            echo $htmsg;
            echoMsg($htmsg,$theWorkerRedis,true);


            //记录日志
            $info = $serv->connection_info($fd);
            $data = array(
                'type' => 0,
                'deviceid' =>$deviceID,
                'description' => '关闭连接',
                'ip' => $info['remote_ip'],
                'createtime' => time(),
            );

            $data['description'] = "关闭连接".$htmsg;
            $theDb->insert('devicelog', $data);
        }

//        $serv->db[$serv->worker_id]->update('device', array('cfglastouttime' => time(), 'cfgfd' => 0), array('id' => $this->client->get($fd, 'id')));
        $theWorkerRedis->del('dtheart:'.$fd);
        $this->id2fd->del($this->client->get($fd, 'deviceID'));
        $this->client->del($fd);
        $this->heartlist->del($fd);

    }


    // 4. 正常结束Server的时候，首先结束Worker进程 kill -10
    public function onWorkerStop($server,$worker_id)
    {
        echo "\n WorkerStop[$worker_id]|pid=".posix_getpid().".\n";
        $server->workerdb=null;
        unset( $server->redis[$worker_id]);
        unset( $server->db[$worker_id]);

    }
	
	
	/**
	 * @param Server $serv
	 * @param $fd
	 * @param $data
	 * @return mixed
	 */
    public function sendtoclient($serv,$fd,$data)
    {
        $ret=$serv->send($fd, $data);
        if($ret){
            $this->client->incr($fd,'stcount', 1);
            $this->client->incr($fd,'sentcts', strlen($data)/2);
        } 

		// 探霸调试发送数据调试数据
		/* else {
	        $deviceId = $this->client->get($fd, 'deviceID');
	        
	        if ($deviceId) {
		        Log::dataLog($deviceId)->info('发送数据给探霸|', [
		        	'fd' => $fd,
		        	'pid' => getmypid(),
		        	// 获取swoole的发送失败错误
			        'error' => $serv->getLastError(),
			        'exists' => $serv->exist($fd),
			        // 连接信息
			        'info' => $serv->connection_info($fd)
		        ]);
	        }
        } */
        
        // 记录向发探霸发送数据信息
	   
        return $ret;

    }

    public function sentMsgOut($serv,$sttdata,$theRedis)
    {
        $theRedis->publish(getDbNum() . REDIS_TB_STATUS_SERVER, json_encode($sttdata));

        $array_hmset_date = array(
            'fd' => $sttdata['fd'],
            'updatetime' => $sttdata['time'],
            'status' => $sttdata['status']
        );
        $theRedis->hMset($sttdata['deviceid'], $array_hmset_date);

    }
    /**
     * 数据库写入操作投递到任务队列中完成,只能在worker进程中使用
     * @param $serv
     * @param $table
     * @param $param
     * @param array $where
     * @param string $option
     * Created by: xiebaoxin
     */
    public function sentmysqltask($serv,$table,$param,$where=[],$option='insert')
    {

        if(isset($table) && isset($option) && is_array($param) ){
            //  $serv->task(json_encode(['option'=>$option,'table'=>$table,'param'=>$param,'where'=>$where]));

            $srarr=$this->swoole_serialize::pack(['option'=>$option,'table'=>$table,'param'=>$param,'where'=>$where]);

            $serv->task($srarr);
        }

        return ;
    }


    /**
     * 发送监听反馈
     * @param $serv
     * @param $wskt_data
     * Created by: xiebaoxin
     */
    public function sentMsgtoTcp($serv,$wbsktdata,$defcmd='getdata')
    {
    	//dump($wbsktdata);
	    if ($defcmd === 'getsecrar') {
		    $this->pushDataToRedis($wbsktdata, $serv->worker_id);
	    }
	    // $this->pushDataToRedis($wbsktdata, $serv->worker_id);
        foreach ($this->listener as $row) {
            if($serv->exist($row['fd']) && isset($row['cmddata'])){
                    if($row['cmddata']==$defcmd){                    //获取监听端口列表需要验证
//        echo "{$wbsktdata['tbmac']}+".$wbsktdata['type'].PHP_EOL;
                        $serv->send($row['fd'],json_encode($wbsktdata).PHP_EOL);
                    }
            }
        }

    }
	
	/**
	 * 将获取到的数据实时放入redis中
	 * @param $data
	 * @param $workerId
	 */
    protected function pushDataToRedis($data, $workerId)
    {
	    /**
	     * @var \Redis $redis
	     */
        $redis = TbRedis::init($workerId);
        
        // 只处理对应的数据
	    //$type = new \stdClass();
        if (in_array($data['type'], TYPE_LISTS)) {
            //getRedisTypeKey($data['type']);
	
            
	        /**
	         * @var \TYPE $type
	         */
            /*$type->find     = $data['find'];
            $type->time     = $data['time'];
            $type->mac      = $data['mac'];
            $type->rssi     = $data['rssi'];
            $type->json     = $data['json'] ?? '';
            $type->tbId     = $data['tbid'];
            $type->devId    = $data['devid'];
            $type->devKey   = $data['devkey'];
            $type->tbMac    = $data['tbmac'];
            $type->sFind    = $data['sfind'] ?? '';*/
//            switch ($data['type']){
//	            case TYPE_AP:
//		            /**
//		             * @var \AP $type
//		             */
//		            $type->channel      = $data['channel'];
//		            $type->frameType    = $data['frametype'];
//		            $type->ssid         = $data['ssid'];
//		            $type->ssidLen      = $data['ssid_len'];
//		            $type->isHidden     = $data['is_hidden'];
//	            	break;
//	            case TYPE_RF:
//		            /**
//		             * @var \RF $type
//		             */
//		            $type->battery      = $data['battery'];
//		            $type->sendRSSI     = $data['sendRssi'];
//		            $type->rssi         = $data['rssi'];
//		            $type->urgentStatus = $data['urgentStatus'];
//	            	break;
//	            case TYPE_STA:
//		            /**
//		             * @var \STA $type
//		             */
//		            $type->hasAp        = $data['Has_ap'];
//		            $type->frameType    = $data['frametype'];
//		            $type->apMac        = $data['ApMac'];
//	            	break;
//            }
            
            $redis->lPush(getRedisTypeKey($data['type']), json_encode($data));
        }
    }


    public function getpackmsg($fd)
    {
        $deviceID=$this->client->get($fd, 'deviceID');
        $htmsg=PHP_EOL;

        $htmsg .="=== heart sent/losed={$this->client->get($fd,'hrtsd')}/{$this->client->get($fd,'hrtrsv')} ". date('Y-m-d H:i:s') . " ===\n";
        $htmsg .=  "===*** fd({$fd})[{$deviceID}] recieve  pack total ".$this->client->get($fd,'tcount').",".$this->client->get($fd,'receivects')." Bit;";
        $htmsg .=  "sent  pack total ".$this->client->get($fd,'stcount').",".$this->client->get($fd,'sentcts')." Bit;";
        $htmsg .="invalid [".$this->client->get($fd,'invalid')."],";

        $htmsg .="ap[".$this->client->get($fd,'a000dc')."],";
        $htmsg .="sta[".$this->client->get($fd,'a001dc')."],";
        $htmsg .="rf[".$this->client->get($fd,'a002dc')."],";
        $htmsg .="errotype[".$this->client->get($fd,'Errotc')."]";

        $htmsg .="***===".PHP_EOL;
        return $htmsg;
    }


    /**
     * onListenerConnect
     * @param $serv
     * @param $fd
     * Created by: xiebaoxin
     * Date: ${DATE}
     * Time: ${TIME}
     */
    public function onListenerConnect($serv, $fd, $from_id)
    {
        $info = $serv->connection_info($fd);
        $serv->protect($fd);
        echo " \n Listener[$fd]Connect : fd=$fd, from={$info['remote_ip']}:{$info['remote_port']}..worker_id=".$serv->worker_id."\n";
        //   var_dump($info);
        if(isset($info) ){
            $this->listener->set($fd, array( 'fd' => $fd,'remote_ip' => $info['remote_ip'],'remote_port' =>$info['remote_port'],'lasttime' => time()));

        }
    }

    /***
     * 监听端口接收数据
     * onListenerReceive
     * @param $serv
     * @param $fd
     * @param $form_id
     * @param $data  命令参数
     * Created by: xiebaoxin
     * Date: 20171011
     */
    public function onListenerReceive($serv, $fd, $form_id, $data)
    {
        echo "\n =={$fd}== listenner :".json_encode($data)." === ".PHP_EOL;
        if(isset($data) && !empty($data) ){
            $data=json_decode($data,true);
            if(isset($data['command'])) {
                $this->listener->set($fd, array('cmddata' =>$data['command'],'lasttime' => time()));
            }

            $serv->send($fd, "command is ok \n");
        }
    }


    public function onListenerClose($serv, $fd, $reactorId)
    {
        echo " \n Listener[$fd]Close \n";
        $this->listener->del($fd);
    }


    // 6. 最后回收Master进程
    public function onShutdown( $serv )
    {

    }

    /**
     * 析构方法
     */
    public function __destruct()
    {
       echo  stdLog::print("set all device offline successful");
    }
}
